package bma.common.netty.protobuf.core;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBufferInputStream;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;

import bma.common.langutil.ai.AIUtil;
import bma.common.langutil.ai.common.AITimeoutException;
import bma.common.langutil.ai.stack.AIStack;
import bma.common.langutil.ai.stack.AIStackROOT;
import bma.common.langutil.concurrent.ProcessTimerTaskAbstract;
import bma.common.langutil.concurrent.TimerManager;
import bma.common.netty.protobuf.core.BaseProtos.PError;
import bma.common.netty.protobuf.core.BaseProtos.PHeader;
import bma.common.netty.protobuf.core.BaseProtos.PHeader.MessageType;
import bma.common.netty.protobuf.core.BaseProtos.PRequest;
import bma.common.netty.protobuf.core.impl.PClientNetty;
import bma.common.netty.protobuf.core.impl.PRequestContextNetty;
import bma.common.netty.protobuf.core.impl.PResponseContextImpl;

/**
 * 处理Protobuf请求
 * 
 * @author guanzhong
 * 
 */
public class NCHProtobufService extends SimpleChannelUpstreamHandler {

	final org.slf4j.Logger log = org.slf4j.LoggerFactory
			.getLogger(NCHProtobufService.class);

	protected NettyProtobufConfig config;
	protected PDispatcher dispatcher;

	protected AtomicInteger msgId = new AtomicInteger();
	protected Map<Integer, AIStack<PResponseContext>> await = new ConcurrentHashMap<Integer, AIStack<PResponseContext>>();

	public NCHProtobufService() {
		super();
	}

	public NCHProtobufService(NettyProtobufConfig cfg, PDispatcher disp) {
		super();
		this.config = cfg;
		this.dispatcher = disp;
	}

	public PDispatcher getDispatcher() {
		return dispatcher;
	}

	public void setDispatcher(PDispatcher dispatcher) {
		this.dispatcher = dispatcher;
	}

	public NettyProtobufConfig getConfig() {
		return config;
	}

	public void setConfig(NettyProtobufConfig config) {
		this.config = config;
	}

	@Override
	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
			throws Exception {
		Object obj = e.getMessage();
		if (!(obj instanceof ChannelBuffer)) {
			ctx.sendUpstream(e);
			return;
		}

		ChannelBuffer buffer = (ChannelBuffer) obj;
		ChannelBufferInputStream in = new ChannelBufferInputStream(buffer);

		PHeader header = null;
		try {
			header = PHeader.parseDelimitedFrom(in);
		} catch (Exception x) {
			log.error("Error occurred during parsing message.", x);
			return;
		}

		if (log.isDebugEnabled()) {
			log.debug("pheader - {}", PUtil.toString(header));
		}

		if (!header.getType().equals(MessageType.E_REQUEST)) {
			// response or error
			int id = header.getId();
			final AIStack<PResponseContext> stack = await.remove(id);
			if (stack != null) {
				final PHeader fheader = header;
				final InputStream fin = in;
				config.getExecutor().execute(new Runnable() {

					@Override
					public void run() {
						if (fheader.getType().equals(MessageType.E_ERROR)) {
							PError perr;
							try {
								perr = PError.parseDelimitedFrom(fin);
							} catch (IOException e) {
								AIUtil.safeFailure(stack, e);
								return;
							}
							PErrorException ex = PErrorException.create(perr);
							AIUtil.safeFailure(stack, ex);
						} else {
							PResponseContextImpl response = new PResponseContextImpl(
									fheader, fin);
							AIUtil.safeSuccess(stack, response);
						}
					}
				});

			} else {
				if (log.isDebugEnabled()) {
					log.debug("skip response - [{}]", id);
				}
			}
			return;
		}

		// request
		PRequest req = PRequest.parseDelimitedFrom(in);
		if (log.isDebugEnabled()) {
			log.debug("[{}] prequest - {}", header.getId(), PUtil.toString(req));
		}

		final PRequestContext rctx = new PRequestContextNetty(ctx.getChannel(),
				header, req, in) {

			@Override
			public void response(Object msg, Throwable err) {
				sendResponse(channel, header, msg, err);
			}

			@Override
			public PClient getClient() {
				return getRemoteClient(channel);
			}

		};
		AIStack<PService> stack = new AIStackROOT<PService>() {

			@Override
			public boolean end(final PService result, Throwable t) {
				if (t != null) {
					rctx.response(null, t);
				} else {
					if (result == null) {
						if (log.isWarnEnabled()) {
							log.warn("handle service not found - {}", rctx);
						}
						PError msg = PUtil.error404("'"
								+ rctx.getRequest().getTarget()
								+ "' service not found");
						rctx.response(msg, null);
					} else {
						config.getExecutor().execute(new Runnable() {

							@Override
							public void run() {
								try {
									result.process(rctx);
								} catch (Exception e) {
									if (log.isWarnEnabled()) {
										log.warn("handle service error - "
												+ rctx, e);
									}
									rctx.response(null, e);
								}
							}
						});

					}
				}

				return true;
			}

		};
		dispatcher.dispatch(stack, rctx);
	}

	@Override
	public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e)
			throws Exception {
		super.channelClosed(ctx, e);
		List<AIStack<PResponseContext>> tmp = new ArrayList<AIStack<PResponseContext>>(
				await.values());
		await.clear();
		Throwable t = new IOException("closed");
		for (AIStack<PResponseContext> stack : tmp) {
			AIUtil.safeFailure(stack, t);
		}
	}

	public int sendRequest(Channel ch, AIStack<PResponseContext> stack,
			PRequest req, Object data, int timeout) {
		return sendRequest(ch, stack, req, data, timeout, false);
	}

	public int sendRequest(Channel ch, AIStack<PResponseContext> stack,
			PRequest req, Object data, int timeout, boolean closed) {
		int id = msgId.incrementAndGet();
		while (await.containsKey(id)) {
			id = msgId.incrementAndGet();
		}
		PHeader header = PHeader.newBuilder().setType(MessageType.E_REQUEST)
				.setVersion(config.getProtocolVersion()).setId(id).build();
		try {
			PUtil.write(ch, header, req, data, closed);
		} catch (Exception e) {
			if (log.isDebugEnabled()) {
				log.debug("write fail,{},{}", header, data);
			}
			if (stack != null) {
				AIUtil.safeFailure(stack, e);
			}
			return id;
		}

		if (stack != null) {
			await.put(id, stack);
		}
		if (timeout > 0) {
			final int fid = id;
			ProcessTimerTaskAbstract task = new ProcessTimerTaskAbstract(
					System.currentTimeMillis() + timeout, true) {

				@Override
				public void run() {
					final AIStack<PResponseContext> stack = await.remove(fid);
					if (stack != null) {
						config.getExecutor().execute(new Runnable() {
							@Override
							public void run() {
								Throwable t = new AITimeoutException();
								AIUtil.safeFailure(stack, t);
							}
						});
					}
				}
			};
			TimerManager.checkInstance(config.getTimer()).postTimerTask(task);
		}
		return id;
	}

	public PClient getRemoteClient(Channel ch) {
		return new PClientNetty(ch) {

			@Override
			public int write(AIStack<PResponseContext> stack, PRequest request,
					Object data, int timeout) {
				return sendRequest(channel, stack, request, data, timeout);
			}

			@Override
			public void writeAndClose(PRequest request, Object data) {
				sendRequest(channel, null, request, data, 0, true);
			}

			@Override
			public void discard(int id) {
				await.remove(id);
			}
		};
	}

	public void sendResponse(Channel ch, PHeader requestHeader, Object data,
			Throwable t) {
		if (!requestHeader.getType().equals(MessageType.E_REQUEST)) {
			throw new IllegalArgumentException("invalid response "
					+ requestHeader);
		}
		if (t != null) {
			if (t instanceof PErrorException) {
				PErrorException perr = (PErrorException) t;
				data = perr.toError();
			} else {
				data = PError
						.newBuilder()
						.setCode(500)
						.setType(t.getClass().getName())
						.setMessage(
								t.getMessage() == null ? "" : t.getMessage())
						.build();
			}
		}
		PHeader responseHeader = PUtil.responseHeader(requestHeader,
				data instanceof PError ? MessageType.E_ERROR
						: MessageType.E_RESPONSE);
		PUtil.write(ch, responseHeader, null, data, false);
	}

}
